package com.example;

import org.apache.pulsar.client.api.*;
import org.apache.pulsar.client.impl.schema.StringSchema;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

/**
 * @author: linfeng
 * @date: 2021/11/12 18:47
 */
public class SimpleProducer {

    private static PulsarClient client;

    private static final String TOPIC_NAME = "TEST-TOPIC";

    public static void main(String[] args) throws PulsarClientException, InterruptedException {
        client = PulsarClient.builder()
                .serviceUrl("pulsar://localhost:6650")
                .build();

        produceMessage();
    }


    private static void produceMessage() throws PulsarClientException {
        Producer<String> stringProducer = client.newProducer(Schema.STRING)
                .topic(TOPIC_NAME)
                .create();

        CompletableFuture<MessageId> future = stringProducer.sendAsync("hello world");
        future.whenComplete((messageId, throwable) -> {
            System.out.println("消息已发送，messageId：" + messageId);
            if(throwable != null){
                throwable.printStackTrace();
            }
            try {
                stringProducer.close();
            } catch (PulsarClientException e) {
                e.printStackTrace();
            }
        });
    }


}
